﻿using Brc_PaaS.Common.Model.RabbitMq;
using Microsoft.Extensions.Configuration;
using RabbitMQ.Client;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace Brc_PaaS.Core.RabbitMq
{
    public class RabbitMQConnectionPool
    {
        private readonly ConcurrentQueue<IConnection> FreeConnectionQueue;//空闲连接对象队列
        private readonly ConcurrentDictionary<IConnection, bool> BusyConnectionDic;//使用中（忙）连接对象集合
        private readonly ConcurrentDictionary<IConnection, int> MQConnectionPoolUsingDicNew;//连接池使用率
        private readonly RabbitMQSettingEntity RabbitMQSetting;//MQ配置实体
        private readonly Semaphore MQConnectionPoolSemaphore;
        private readonly object freeConnLock = new object(), addConnLock = new object();
        public const int DefaultMaxConnectionUsingCount = 10000;//默认最大连接可访问次数

        public RabbitMQConnectionPool(RabbitMQSettingEntity rabbitMQSetting)
        {
            try
            {
                RabbitMQSetting = rabbitMQSetting;
                FreeConnectionQueue = new ConcurrentQueue<IConnection>();
                BusyConnectionDic = new ConcurrentDictionary<IConnection, bool>();
                MQConnectionPoolUsingDicNew = new ConcurrentDictionary<IConnection, int>();//连接池使用率
                MQConnectionPoolSemaphore = new Semaphore(RabbitMQSetting.MaxSize, RabbitMQSetting.MaxSize);//信号量，控制同时并发可用线程数
            }
            catch (Exception ex)
            {
                throw new Exception($"RabbitMQConnectionPool初始化失败,异常信息:{ex.Message}");
            }
        }

        public IConnection CreateMQConnection()
        {
            var factory = CrateFactory();
            var connection = factory.CreateConnection();
            return connection;
        }

        private ConnectionFactory CrateFactory()
        {
            var factory = new ConnectionFactory()
            {
                HostName = RabbitMQSetting.ServerAddress,
                //Port = RabbitMQSetting.ServerPort,
                VirtualHost = RabbitMQSetting.VirtualHost,
                UserName = RabbitMQSetting.UserName,
                Password = RabbitMQSetting.Password,
                RequestedHeartbeat = 60, //心跳超时时间
                AutomaticRecoveryEnabled = true //自动重连
            };
            return factory;
        }


        public IConnection CreateMQConnectionInPoolNew()
        {
            MQConnectionPoolSemaphore.WaitOne(10000);//当<MaxConnectionCount时，会直接进入，否则会等待直到空闲连接出现
            IConnection mqConnection = null;
            try
            {
                if (FreeConnectionQueue.TryDequeue(out mqConnection)) //如果存在空闲连接
                {
                    if (MQConnectionPoolUsingDicNew[mqConnection] + 1 < DefaultMaxConnectionUsingCount && mqConnection.IsOpen) //判断使用次数及连接状态是否可以使用
                    {
                        BusyConnectionDic[mqConnection] = true;//加入到忙连接集合中
                        MQConnectionPoolUsingDicNew[mqConnection] = MQConnectionPoolUsingDicNew[mqConnection] + 1;//使用次数加1
                        return mqConnection;
                    }
                    else
                    {
                        if (mqConnection.IsOpen)
                        {
                            mqConnection.Close();
                        }
                        mqConnection.Dispose();
                    }
                }
                if (FreeConnectionQueue.Count + BusyConnectionDic.Count < RabbitMQSetting.MaxSize)//如果已有连接数小于最大可用连接数，则直接创建新连接
                {
                    lock (addConnLock)
                    {
                        if (FreeConnectionQueue.Count + BusyConnectionDic.Count < RabbitMQSetting.MaxSize)
                        {
                            mqConnection = CreateMQConnection();
                            BusyConnectionDic[mqConnection] = true;//加入到忙连接集合中
                            MQConnectionPoolUsingDicNew[mqConnection] = 1;
                            return mqConnection;
                        }
                    }
                }

                if (!FreeConnectionQueue.TryDequeue(out mqConnection)) //如果没有可用空闲连接，则重新进入等待排队
                {
                    return CreateMQConnectionInPoolNew();
                }
                else if (MQConnectionPoolUsingDicNew[mqConnection] + 1 > DefaultMaxConnectionUsingCount || !mqConnection.IsOpen) //如果取到空闲连接，判断是否使用次数是否超过最大限制,超过则释放连接并重新创建
                {
                    if (mqConnection.IsOpen)
                    {
                        mqConnection.Close();
                    }
                    mqConnection.Dispose();
                    mqConnection = CreateMQConnection();
                    MQConnectionPoolUsingDicNew[mqConnection] = 0;
                }

                BusyConnectionDic[mqConnection] = true;//加入到忙连接集合中
                MQConnectionPoolUsingDicNew[mqConnection] = MQConnectionPoolUsingDicNew[mqConnection] + 1;//使用次数加1
                return mqConnection;
            }
            catch (Exception ex)
            {
                if (mqConnection != null)
                {
                    ResetMQConnectionToFree(mqConnection);
                }
                else
                {
                    MQConnectionPoolSemaphore.Release();
                }
                throw new Exception($"创建连接失败,失败原因：{ex.Message}");
            }
        }

        public void ResetMQConnectionToFree(IConnection connection)
        {
            try
            {
                lock (freeConnLock)
                {
                    bool result = false;
                    if (!BusyConnectionDic.TryRemove(connection, out result))//从忙队列中取出
                    {
                        if (!BusyConnectionDic.TryRemove(connection, out result))
                        {
                            throw new Exception("释放连接失败");
                        }
                    }
                    if (FreeConnectionQueue.Count + BusyConnectionDic.Count > RabbitMQSetting.MaxSize)//如果因为高并发出现极少概率的>MaxConnectionCount，则直接释放该连接
                    {
                        connection.Close();
                        connection.Dispose();
                    }
                    else if (connection.IsOpen)
                    {
                        FreeConnectionQueue.Enqueue(connection);//加入到空闲队列，以便持续提供连接服务
                    }
                }
            }
            catch (Exception ex)
            {
                throw new Exception($"释放连接失败,失败原因：{ex.Message}");
            }
            finally
            {
                MQConnectionPoolSemaphore.Release();//释放一个空闲连接信号
            }

        }

    }
}
